package com.dcloud.streamweb.sysconfig.service.impl;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dcloud.streamweb.constants.FlinkYarnRestUriConstants;
import com.dcloud.streamweb.constants.SystemConstants;
import com.dcloud.streamweb.enums.DeployModeEnum;
import com.dcloud.streamweb.enums.SysConfigEnum;
import com.dcloud.streamweb.enums.SysConfigEnumType;
import com.dcloud.streamweb.enums.SysErrorEnum;
import com.dcloud.streamweb.exception.BizException;
import com.dcloud.streamweb.sysconfig.entity.SystemConfig;
import com.dcloud.streamweb.sysconfig.entity.dto.SystemConfigDTO;
import com.dcloud.streamweb.sysconfig.mapper.SystemConfigMapper;
import com.dcloud.streamweb.sysconfig.service.ISystemConfigService;
import com.dcloud.streamweb.util.FileUtils;
import com.dcloud.streamweb.util.HttpServiceCheckerUtil;
import com.dcloud.streamweb.util.HttpUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.util.CollectionUtils;

import java.net.URL;
import java.util.Collections;
import java.util.List;

/**
 * @Description: system_config
 * @Author: jeecg-boot
 * @Date:   2024-07-11
 * @Version: V1.0
 */
@Service
@Slf4j
public class SystemConfigServiceImpl extends ServiceImpl<SystemConfigMapper, SystemConfig> implements ISystemConfigService {

    @Autowired
    private SystemConfigMapper systemConfigMapper;

    @Override
    public List<SystemConfigDTO> getSystemConfig(SysConfigEnumType sysConfigEnumType) {
        return SystemConfigDTO.toListDTO(systemConfigMapper.selectAllConfig(sysConfigEnumType != null
                ? sysConfigEnumType.name() : null));
    }

    @Override
    public void addOrUpdateConfigByKey(String key, String value) {
        //检测输入的参数是否有效
        this.checkParam(key, value);

        if (SysConfigEnum.FLINK_HOME.equals(SysConfigEnum.getSysConfigEnum(key))) {
            FileUtils.createSqlHome(value);
        }
        SystemConfig systemConfig = systemConfigMapper.selectConfigByKey(key);
        if (systemConfig == null) {
            systemConfigMapper.insert(new SystemConfig(key, value.trim()));
        } else {
            systemConfigMapper.updateByKey(new SystemConfig(key, value.trim()));
        }
    }

    @Override
    public void deleteConfigByKey(String key) {
        systemConfigMapper.deleteByKey(key);
    }

    @Override
    public String getSystemConfigByKey(String key) {
        List<SystemConfigDTO> list = this.getSystemConfig(null);
        if (CollectionUtils.isEmpty(list)) {
            return null;
        }
        return SystemConfigDTO.toMap(list).get(key);
    }

    /**
     * 功能描述 根据运行模式不同，返回不同的flinkhttp地址
     *
     * @param deployModeEnum
     * @return java.lang.String
     * @author shijf
     * @date 2024/7/24 下午11:50
     */
    @Override
    public String getFlinkHttpAddress(DeployModeEnum deployModeEnum) {

        switch (deployModeEnum) {
            //单机类型
            case LOCAL:
                String urlLocal = this.getSystemConfigByKey(SysConfigEnum.FLINK_REST_HTTP_ADDRESS.getKey());
                if (StringUtils.isEmpty(urlLocal)) {
                    throw new BizException(SysErrorEnum.SYSTEM_CONFIG_IS_NULL_FLINK_REST_HTTP_ADDRESS);
                }
                if (HttpServiceCheckerUtil.checkUrlConnect(urlLocal)) {
                    return urlLocal.trim();
                }
                throw new BizException("网络异常 url=" + urlLocal);

            //standalone 多集群类型
            case STANDALONE:
                String urlHA = this.getSystemConfigByKey(SysConfigEnum.FLINK_REST_HA_HTTP_ADDRESS.getKey());
                if (StringUtils.isEmpty(urlHA)) {
                    throw new BizException(SysErrorEnum.SYSTEM_CONFIG_IS_NULL_FLINK_REST_HA_HTTP_ADDRESS);
                }
                String[] urls = urlHA.split(SystemConstants.SEMICOLON);
                for (String http : urls) {
                    if (HttpServiceCheckerUtil.checkUrlConnect(http)) {
                        return http.trim();
                    }
                }
                throw new BizException("网络异常 url=" + urlHA);
            default:
                throw new BizException("不支持该模式");
        }
    }

    @Override
    public String getUploadJarsPath() {
        String path = this
                .getSystemConfigByKey(SysConfigEnum.FLINK_STREAMING_PLATFORM_WEB_HOME.getKey());
        if (StringUtils.isEmpty(path)) {
            throw new BizException("请先去系统设置界面设置Flink管理平台目录(即flink_streaming_platform_web)");
        }
        return path + SystemConstants.VIRGULE + SystemConstants.JAR_ROOT_PATH;
    }

    @Override
    public String getFlinkAddress(DeployModeEnum deployModeEnum) {
        try {
            String url = this.getFlinkHttpAddress(deployModeEnum);
            URL address = new URL(url);
            String host = address.getHost();
            Integer port = address.getPort() == -1 ? 80 : address.getPort();
            return host + ":" + port;
        } catch (Exception e) {
            log.error("getFlinkAddress is error", e);
        }
        return null;
    }

    @Override
    public String getYarnRmHttpAddress() {
        String urlHa = this.getSystemConfigByKey(SysConfigEnum.YARN_RM_HTTP_ADDRESS.getKey());
        if (StringUtils.isEmpty(urlHa)) {
            throw new BizException(SysErrorEnum.SYSTEM_CONFIG_IS_NULL_YARN_RM_HTTP_ADDRESS);
        }

        return getActiveYarnUrl(urlHa);
    }


    private void checkParam(String key, String value) {
        if (StringUtils.isEmpty(key) || StringUtils.isEmpty(value)) {
            throw new BizException(SysErrorEnum.PARAM_IS_NULL);
        }
        SysConfigEnum sysConfigEnum = SysConfigEnum.getSysConfigEnum(key);

        //参数必须要 “/” 结尾
        if (SysConfigEnum.YARN_RM_HTTP_ADDRESS.equals(sysConfigEnum)
                || SysConfigEnum.FLINK_REST_HTTP_ADDRESS.equals(sysConfigEnum)
                || SysConfigEnum.FLINK_REST_HA_HTTP_ADDRESS.equals(sysConfigEnum)) {
            if (!StrUtil.endWith(value, SystemConstants.SLASH)) {
                throw new BizException("必须以/结尾");
            }
            if (!StrUtil.startWith(value, SystemConstants.HTTP_KEY)) {
                throw new BizException("必须以http或者https开头");
            }
        }
        if (SysConfigEnum.DINGDING_ALARM_URL.equals(sysConfigEnum)) {
            if (!StrUtil.startWith(value, SystemConstants.HTTP_KEY)) {
                throw new BizException("必须以http或者https开头");
            }
        }

        this.checkUrlValid(sysConfigEnum, value);

        if (SysConfigEnum.FLINK_HOME.equals(sysConfigEnum)) {
            if (!StrUtil.endWith(value, SystemConstants.SLASH)) {
                throw new BizException("必须以/结尾");
            }
            if (!StrUtil.startWith(value, SystemConstants.SLASH)) {
                throw new BizException("必须以/开头");
            }
        }
        if (SysConfigEnum.FLINK_STREAMING_PLATFORM_WEB_HOME.equals(sysConfigEnum)) {
            if (!StrUtil.startWith(value, SystemConstants.SLASH)) {
                throw new BizException("必须以/开头");
            }
            if (!StrUtil.endWith(value, SystemConstants.SLASH)) {
                throw new BizException("必须以/结尾");
            }

        }
    }

    private void checkUrlValid(SysConfigEnum sysConfigEnum, String url) {
        switch (sysConfigEnum) {
            case FLINK_REST_HTTP_ADDRESS:
            case DINGDING_ALARM_URL:
                if (!HttpServiceCheckerUtil.checkUrlConnect(url)) {
                    throw new BizException("网络异常 url=" + url);
                }
                break;
            case YARN_RM_HTTP_ADDRESS:
            case FLINK_REST_HA_HTTP_ADDRESS:
                String[] urls = url.split(SystemConstants.SEMICOLON);
                for (String http : urls) {
                    if (!HttpServiceCheckerUtil.checkUrlConnect(http)) {
                        throw new BizException("网络异常 url=" + http);
                    }
                }
                break;
            default:
                break;
        }

    }

    private String getActiveYarnUrl(String urlHa) {
        String[] urls = urlHa.split(SystemConstants.SEMICOLON);
        for (String http : urls) {
            try {
                String url = HttpUtil
                        .buildUrl(http, FlinkYarnRestUriConstants.URI_YARN_INFO);
                String request = cn.hutool.http.HttpUtil.get(url, HttpServiceCheckerUtil.TIMEOUTMILLSECONDS);
                if (StringUtils.isNotEmpty(request)) {
                    JSONObject jsonObject = (JSONObject) JSON.parse(request);
                    String haState = jsonObject.getJSONObject("clusterInfo").get("haState").toString();
                    if ("ACTIVE".equalsIgnoreCase(haState)) {
                        return http;
                    }
                }
            } catch (Exception e) {
                log.error("单个http异常={}", http, e);
            }
        }
        throw new BizException("网络异常 url=" + urlHa);
    }
}
